package com.bsj.power.common.config.rocketmq.producer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.atomic.LongAdder;

/**
 * @author yinhao
 * @version 1.0
 * @description RocketMq消息发送类
 * @date 2024/6/14
 */
@Slf4j
@Component
public class BaseMqPushUtil {

    @Autowired
    private DefaultMQProducer defaultMQProducer;

    /**
     * 发送成功的消息数量
     */
    private LongAdder msgSerNum = new LongAdder();

    /**
     * 根据主题 tag 发送消息
     * @param bytes
     * @param tag
     */
    public void sendTopicBytes(String topic,  String tag, byte[] bytes) {
        //消费者的大tag和小tag和数据
        Message msg = new Message(topic, tag, bytes);
        try {
            send(msg);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 内部真正的发送
     * @param msg
     * @throws Exception
     */
    private void send(Message msg) throws Exception {
        SendResult result = defaultMQProducer.send(msg);
        if (!SendStatus.SEND_OK.equals(result.getSendStatus())) {
            throw new Exception("MQ发送失败");
        }
        String messageId = result.getMsgId();
        msgSerNum.increment();
        log.info("消息发送成功数量：{},消息ID：{},消息Topic： {}, 消息tag： {}, 发送的值：{}",msgSerNum.sum(), messageId,
                msg.getTopic(), msg.getTags(), new String(msg.getBody()));
    }
}
